package org.ucas.faker.web.fetcher;

import org.ucas.faker.web.fetcher.cond.FetchCond;
import org.ucas.faker.web.fetcher.meta.JoinTable;
import org.ucas.faker.web.fetcher.meta.Table;
import org.ucas.faker.web.fetcher.op.Join;
import org.ucas.faker.web.fetcher.parse.JoinGraph;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;

public class Joiner implements Serializable{

  Join target;

  final JoinGraph graph;

  public Joiner(JoinGraph graph) {
    this.graph = graph;
  }

  Set<Table> joined = new HashSet<>();

  Map<Table,FetchCond> todo = new HashMap<>();

  private Join shape(Join join) {
    if(join ==  null) return null;
    boolean hasJoined = false;
    FetchCond down = null;
    if(joined.contains(join.right)) {
      hasJoined = true;
      down = join.cond;
    } else if(todo.containsKey(join.right)) {
      FetchCond cond = todo.get(join.right);
      join.cond = cond == null? join.cond : cond.and(join.cond);
      todo.remove(join.right);
    }
    joined.add(join.right);

    if(join.left == null) return hasJoined? null:join;
    if(join.left instanceof JoinTable) {
      Join next = ((JoinTable) join.left).join;
      if(down != null) next.cond = down.and(next.cond);
      next = shape(next);
      ((JoinTable) join.left).join = next;
      return hasJoined? next : join;
    } else if(target != null) {
        Join tmp = new Join();
        tmp.left = null;
        tmp.right = join.left;
        if(down != null) tmp.cond = down;
        tmp = shape(tmp);
        join.left = tmp == null? null : new JoinTable(tmp);
        return hasJoined? tmp : join;
    } else return join;
  }

  public void merge(Join join) {
    join = shape(join);
    if(target != null) join.mostLeft().left = new JoinTable(target);
    target = join;
  }

  private static class BFSNode {
    Table table;

    Queue<JoinGraph.JoinEdge> path;

    public BFSNode(Table table, Queue<JoinGraph.JoinEdge> path) {
      this.table = table;
      this.path = path;
    }

    public BFSNode(Table table) {
      this(table, new LinkedBlockingQueue<>());
    }
  }

  private Queue<JoinGraph.JoinEdge> BFS(Table table) {
    Set<Table> visited = new HashSet<>();
    int level = 0;
    Queue<BFSNode> cands = new LinkedBlockingQueue<>();
    cands.add(new BFSNode(table));
    while (!cands.isEmpty() && level < 3) {
      BFSNode node = cands.remove();
      if(node == null) {
        ++level;
        if(!cands.isEmpty()) cands.add(null);
        continue;
      }

      Table cur = node.table;
      visited.add(cur);

      if(joined.contains(cur)) return node.path;

      List<JoinGraph.JoinEdge> edges = graph.edges(cur);
      for(JoinGraph.JoinEdge edge:edges) {
        if(visited.contains(edge)) continue;
        BFSNode child = new BFSNode(edge.target);
        child.path.addAll(node.path);
        child.path.add(edge);
        cands.add(child);
      }
    }
    return null;
  }

  private Join genJoin(Queue<JoinGraph.JoinEdge> path, Table start) {
    if(path == null || start == null || path.isEmpty()) return null;
      JoinGraph.JoinEdge edge = path.remove();
      if(path.isEmpty()) {
        return new Join(null,start,edge.join.cond);
      }
      Join res = new Join();
      res.right = start;
      res.cond = edge.join.cond;
      res.left = new JoinTable(genJoin(path,edge.target));
      return res;
  }

  private Join solution(Table table) {
    if(graph == null) return null;
    Queue<JoinGraph.JoinEdge> path = BFS(table);
    return genJoin(path,table);
  }

  public boolean join(Table table) {
    Join join = solution(table);
    if(join == null) return false;
    merge(join);
    return true;
  }

  public void add(Table table, FetchCond cond) {
    if(joined.contains(table)) {
      if(cond == null) return;
      Join join = target.find(table);
      if(join.cond == null) join.cond = cond;
      else join.cond.and(cond);
    }

    FetchCond conds = todo.getOrDefault(table,null);
    cond = conds == null? cond:conds.and(cond);
    todo.put(table,cond);
  }

  public Table build() {
    if(joined.isEmpty() && todo.size() == 1)
      return todo.keySet().iterator().next();
    int retries = 0;
    while (!todo.isEmpty() && retries < 3) {
      CopyOnWriteArrayList<Table> tables = new CopyOnWriteArrayList(todo.keySet());
      for(Table table:tables)
        if(join(table)) todo.remove(table);
      ++retries;
    }
    if(retries >= 3) throw new DynamicSqlException("can't find right join!");
    return new JoinTable(target);
  }
}
